package com.wudl.flink.instance;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.types.Row;
import org.apache.hadoop.mapreduce.Job;

import java.util.stream.Stream;

/**
 * @ClassName : WriteToMysql
 * @Description : flink-jdbc
 * @Author :wudl
 * @Date: 2021-08-01 20:36
 */

public class WriteToMysql {

    public static void main(String[] args) throws Exception {
        System.out.println("----------args" + args);
        System.out.println("----------args" + args.toString());
        for (int i = 0; i < args.length; i++) {
            System.out.println("-----i---" + args[i]);
        }

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        System.out.println("------------------------");
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        ParameterTool parameter = ParameterTool.fromSystemProperties();
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("888884545555" + parameter.toString());
        System.out.println("fromSystemProperties--------" + parameter.get("programType"));
        System.out.println("fromSystemProperties2222-----------" + parameter.get("java"));
        System.out.println("fromSystemProperties2222-333----------" + parameter.get("mainClass"));
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("88888" + parameter.getNumberOfParameters());
        System.out.println("88888" + parameter.getNumberOfParameters());

        System.out.println("打印参数----" + parameterTool.toString());
        System.out.println("********111**********" + parameterTool.get("java"));
        System.out.println("********111programType**********" + parameterTool.get("programType"));
        System.out.println("********111resourceList**********" + parameterTool.get("resourceList"));
        System.out.println("*******2222***********" + parameterTool.getProperties().getProperty("java"));
        System.out.println("*******3333***********" + parameterTool.getProperties().get("java"));
        System.out.println("*******3333***********" + parameterTool.getProperties().toString());
        System.out.println("*******3333***********" + parameterTool.getProperties());
        System.out.println("*******3333***********" + parameterTool.getProperties());

        System.out.println("////////////////////////////////////////////////////////");
//        env.getConfig().setGlobalJobParameters(parameterTool);

        DataSet<Row> inputMysql = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
                //配置数据库连接信息
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://192.168.1.162:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery("select * from testtable")
                //设置查询的列的类型，根据实际情况定
                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                .finish());


        inputMysql.print();
        System.out.println("************");
//
//        inputMysql.map(new MapFunction<Row, String>() {
//            @Override
//            public String map(Row value) throws Exception {
//                System.out.println(value);
//                return value.toString();
//            }
//        }).print();

        //mysql连接
        String driverClass = "com.mysql.jdbc.Driver";
        String dbUrl = "jdbc:mysql://192.168.1.162:3306/test?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true ";
        String userNmae = "root";
        String passWord = "123456";
        String sql = "insert into testtable02 (id,mamer) values (?,?)";
        //将数据写入mysql数据库
        inputMysql.output(JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://192.168.1.162:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery(sql)
                .finish());
        //这个必须要执行，否则看不到结果
        Job job = Job.getInstance();
        env.execute("insert into mysql");
    }

//    class FilterGenreWithGlobalEnv extends RichFilterFunction<Tuple3<Long, String, String>> {
//        @Override
//        public boolean filter(Tuple3<Long, String, String> movie) throws Exception {
//            String[] genres = movie.f2.split(",");
//            // Get global parameters
//            ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
//            // Read parameter
//            String genre = parameterTool.get("genre");
//            return Stream.of(genres).anyMatch(g -> g.equals(genre));
//        }
//    }

}